package cn.gyn.gynrpc.core.consumer;

import cn.gyn.gynrpc.core.api.*;
import cn.gyn.gynrpc.core.consumer.http.OkHttpInvoker;
import cn.gyn.gynrpc.core.governance.SlidingTimeWindow;
import cn.gyn.gynrpc.core.meta.InstanceMeta;
import cn.gyn.gynrpc.core.util.MethodUtils;
import lombok.extern.slf4j.Slf4j;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.net.SocketTimeoutException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

import static cn.gyn.gynrpc.core.util.TypeUtils.castMethodResult;

/**
 * 消费端动态代理处理类
 * 在动态代理中模拟Http请求调用provider
 */
@Slf4j
public class GynInvocationHandler implements InvocationHandler {
    Class<?> service;
    RpcContext context;
    final List<InstanceMeta> providers;
    //被隔离的providers
    List<InstanceMeta> isolateProviders = new ArrayList<>();
    //半开的providers
    final List<InstanceMeta> halfOpenProviders = new ArrayList<>();

    //统计出错的状态（得看什么才算故障：比如单位时间内出现多少故障才算，比如30秒内出现10次异常，这里是滑动时间窗口（作用是统计每秒发生的次数））
    Map<String, SlidingTimeWindow> windows = new HashMap<>();

    HttpInvoker httpInvoker;

    ScheduledExecutorService executor;

    public GynInvocationHandler(Class<?> clazz, RpcContext context, List<InstanceMeta> providers) {
        this.service = clazz;
        this.context = context;
        //this.context = context.clone();//目前context里都是公用的参数，如果需要创建单独的配置实例，用clone去创建即可
        this.providers = providers;
        int timeout = Integer.parseInt(context.getParameters().getOrDefault("consumer.timeout", "1000"));
        this.httpInvoker = new OkHttpInvoker(timeout);
        //创建一个单线程
        this.executor = Executors.newScheduledThreadPool(1);
        //延迟10秒执行，每60秒执行一次（生产一般是5分钟）
        int halfOpenInitialDelay = Integer.parseInt(context.getParameters()
                .getOrDefault("consumer.halfOpenInitialDelay", "10000"));
        int halfOpenDelay = Integer.parseInt(context.getParameters()
                .getOrDefault("consumer.halfOpenDelay", "60000"));
        this.executor.scheduleWithFixedDelay(this::halfOpen, halfOpenInitialDelay,
                halfOpenDelay, TimeUnit.MILLISECONDS);
        //this.executor.scheduleWithFixedDelay(this::halfOpen, 10, 60, TimeUnit.SECONDS);
    }

    /**
     * 半开探活：将隔离的provider放到半开的halfOpenProviders里
     */
    private void halfOpen() {
        log.debug(" =====> half open isolatedProviders:" + isolateProviders);
        //要先清理，避免重复加
        halfOpenProviders.clear();
        halfOpenProviders.addAll(isolateProviders);
    }

    @Override
    public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
        //这里获取方法名，比较判断object的基础方法，(点进Object,从idea左边三个方块的structure里看)，进行null返回过滤
        String methodName = method.getName();
        if (MethodUtils.checkLocalMethod(methodName)) {
            return null;
        }

        //这里会拦截请求，直接对RpcRrquest进行赋值
        RpcRequest request = new RpcRequest();
        request.setService(service.getCanonicalName());
        request.setMethodSign(MethodUtils.methodSign(method));
        request.setArgs(args);

        //重试2次
//        int retries = Integer.parseInt(context.getParameters().getOrDefault("consumer.retries", "1"));
//        int faultLimit = Integer.parseInt(context.getParameters().getOrDefault("consumer.faultLimit", "10"));
        int retries = context.getConsumerProperties().getRetries();
        int faultLimit = context.getConsumerProperties().getFaultLimit();
        while (retries-- > 0) {
            log.debug("===> retries: " + retries);
            try {
                /*//filer写这里是因为方法只有被调用才需要过滤，因为传参里RpcRequest在这里声明完了，所以过滤器加这里
                // 当Filter接口里只有一个方法的时候才能用lambda表达式，现在有了多个，所以不能用了
                // Filter filter = r -> true;
                if(filter.prefilter(request)){
                    //对于过滤不通过的可以返回空对象，或抛异常给上层处理
                    //还有一种用法是catch的过滤器：如果一个参数上次已经给了，这里直接返回上次调用到的返回值
                    return null;
                }*/

                //foreach方法有副作用:没办法让invoke方法退出,所以这里直接用for循环
                for (Filter filter : this.context.getFilters()) {
                    Object preResult = filter.prefilter(request);
                    //这里用CacheFilter从缓存种读取结果直接返回，不用调provider http请求了。
                    if (preResult != null) {
                        log.debug(filter.getClass().getName() + " ==> preFilter: " + preResult);
                        return preResult;
                    }
                }

                InstanceMeta instance;
                //设置同步块防止并发问题
                synchronized (halfOpenProviders) {
                    //如果需要探活，否则用原逻辑
                    if (halfOpenProviders.isEmpty()) {
                        //List.of创建一个不可变的 List 集合
                        List<InstanceMeta> instances = context.getRouter().route(providers);
                        instance = context.getLoadBalancer().choose(instances);
                        log.debug("loadBalancer.choose(instances) ==> {}", instance);
                    } else {
                        instance = halfOpenProviders.remove(0);
                        log.debug("check alive instance ==> {}", instance);
                    }
                }

                RpcResponse<?> response;
                Object result;
                String url = instance.toUrl();
                //把request变成http请求
                try {
                    response = httpInvoker.post(request, url);
                    log.info("【response】===> {}", response);
                    result = castReturnResult(method, response);
                } catch (Exception e) {
                    //故障的规则统计和隔离
                    //每一次异常，记录一次，统计30s的异常数
                    synchronized (windows) {
                        SlidingTimeWindow window = windows.computeIfAbsent(url, k -> new SlidingTimeWindow());
                        //SlidingTimeWindow window = windows.get(url);
                        window.record(System.currentTimeMillis());
                        log.debug("instance {} in window with {}", url, window.getSum());
                        //发生了10次，就做故障隔离
                        if (window.getSum() >= faultLimit) {
                            isolate(instance);
                        }
                    }

                    throw e;
                }

                synchronized (providers) {
                    //如果是探活成功了（providers里没有此次的实例）
                    if (!providers.contains(instance)) {
                        isolateProviders.remove(instance);
                        providers.add(instance);
                        log.debug("instance {} is recovered, isolateProviders:{}, providers:{}", instance, isolateProviders, providers);
                    }
                }

                //返回结果过滤器
                for (Filter filter : this.context.getFilters()) {
                    //赋值过滤器处理后的 postResponse 给 response
                    Object filterResult = filter.postfilter(request, response, result);
                    if (filterResult != null) {
                        return filterResult;
                    }
                }

                return result;
            } catch (RuntimeException e) {
                //getCause() 方法可以获取引起该异常的原因,如果不是套接字异常，抛出去，否则继续重试
                if (!(e.getCause() instanceof SocketTimeoutException)) {
                    throw e;
                }
            }
        }
        return null;
    }

    /**
     * 隔离方法：将失败的路由从provider里移除掉，记录到一个被隔离的集合里，被隔离的集合用于探活恢复
     * @param instance
     */
    private void isolate(InstanceMeta instance) {
        log.debug(" ===> isolate instance: " + instance);
        providers.remove(instance);
        log.debug(" ===> providers = {}", providers);
        /*if (!isolateProviders.contains(instance)) {
            isolateProviders.add(instance);
            log.debug(" ===> isolateProviders = {}", isolateProviders);
        }*/
        isolateProviders.add(instance);
        log.debug(" ===> isolateProviders = {}", isolateProviders);
    }

    private static Object castReturnResult(Method method, RpcResponse<?> response) {
        //先判断RPC调用是成功还是失败
        if (response.isStatus()) {
            /*//如果是基础数据类型，不需要进行JSON对象转换
            log.debug("返回数据类型" + type);
            if (type.getModule().getName() != null && type.getModule().getName().equals("java.base")) {
                return response.getData();
            }

            //这里的类型和实际调用方需要的类型比如user是不一致的
            //return response.getData();
            JSONObject jsonObject = (JSONObject) response.getData();
            //返回获取结果
            return jsonObject.toJavaObject(type);*/

            //另一种区分基础数据类型的办法
            return castMethodResult(method, response.getData());
        } else {
            RpcException exception = response.getException();
            if (exception != null) {
                log.error("response error.", exception);
                throw exception;
            }
            return null;
            //仅用于检查异常类型，不存储异常对象
            //if (exception instanceof GynrpcException)

            //将异常对象存储在 ex 变量中
            /*if (exception instanceof RpcException ex) {
                //注意这里的exception因为就是Exception类型（父类），因此不能直接抛出，像RuntimeException可以直接抛
                throw ex;
            } else {
                throw new RpcException(exception, RpcException.UnknownEx);
            }*/
        }
    }


}
